package org.kth.dks;

import java.net.InetAddress;
import java.net.URL;
import java.net.UnknownHostException;

import org.apache.log4j.Logger;
import org.kth.dks.dks_comm.ConnectionManager;
import org.kth.dks.dks_comm.DKSNetAddress;
import org.kth.dks.dks_comm.DKSOverlayAddress;
import org.kth.dks.dks_comm.DKSRef;
import org.kth.dks.dks_exceptions.DKSIdentifierAlreadyTaken;
import org.kth.dks.dks_exceptions.DKSNodeAlreadyRegistered;
import org.kth.dks.dks_exceptions.DKSRefNoResponse;
import org.kth.dks.dks_exceptions.DKSTooManyRestartJoins;
import org.kth.dks.dks_marshal.DKSMarshal;
import org.kth.dks.dks_marshal.DKSMessage;
import org.kth.dks.dks_node.DKSNode;
import org.kth.dks.util.AsyncOperation;
import org.kth.dks.util.CommunicationInfo;
import org.kth.dks.util.MessageInfo;
import org.kth.dks.util.NodeInfo;

public class DKSImpl implements DKSInterface, DKSCallbackInterface {

    private static Logger log = Logger.getLogger(DKSImpl.class);  
    
    protected DKSNode myDKSNode = null;
    protected ConnectionManager cm = null;
    protected DKSRef myDKSRef = null;

  public DKSImpl(ConnectionManager _cm, long nodeId, URL nodeAddress)  throws DKSNodeAlreadyRegistered
  {
    cm = _cm;

    myDKSNode = new DKSNode();
    int port = nodeAddress.getPort();
    String ip = null;
    try {
	ip = InetAddress.getByName(nodeAddress.getHost()).getHostAddress();
    }
    catch (UnknownHostException ex) {
	ex.printStackTrace();
	return;
    }

    myDKSRef = DKSRef.newDKSRefByParts( new DKSOverlayAddress(nodeId, 0, 0),
                                       new DKSNetAddress(ip, port) );

    myDKSNode.setConnectionManager(cm, myDKSRef);
  }

  public DKSImpl(ConnectionManager _cm, DKSOverlayAddress over) throws DKSNodeAlreadyRegistered
  {
    cm = _cm;
    myDKSNode = new DKSNode();
    myDKSRef =  DKSRef.newDKSRefByParts( over, new DKSNetAddress(cm.getHostAddress(), cm.getLocalPort()) );
    myDKSNode.setConnectionManager(cm, myDKSRef);
  }

  public DKSImpl(ConnectionManager _cm, long nodeId, long gId, long GUID) throws DKSNodeAlreadyRegistered
  {
    cm = _cm;
    myDKSNode = new DKSNode();
    myDKSRef = DKSRef.newDKSRefByParts( new DKSOverlayAddress(nodeId, gId, GUID),
                                       new DKSNetAddress(cm.getHostAddress(), cm.getLocalPort()) );
    myDKSNode.setConnectionManager(cm, myDKSRef);
  }

  public String getDKSURL() {
    return myDKSRef.toString();
  }

  public void join(long existingnodeId, URL existingnodeAddress ) throws DKSTooManyRestartJoins, DKSIdentifierAlreadyTaken
  {
    DKSRef dest = null;
    if( existingnodeAddress != null ) {
      int port2 = existingnodeAddress.getPort();
      String ip2 = null;
      try {
        ip2 = InetAddress.getByName(existingnodeAddress.getHost()).
            getHostAddress();
      }
      catch (UnknownHostException ex) {
        ex.printStackTrace();
        return;
      }

      dest = DKSRef.newDKSRefByParts( new DKSOverlayAddress(existingnodeId, 0, 0),
                                         new DKSNetAddress(ip2, port2) );
    }

    AsyncOperation joinFuture=null;
    try {
	joinFuture = myDKSNode.newNode( myDKSRef, dest );
        if (joinFuture != null) joinFuture.waitOn();
      } catch (org.kth.dks.dks_exceptions.DKSTooManyRestartJoins ex) {
        throw ex;
      } catch (org.kth.dks.dks_exceptions.DKSIdentifierAlreadyTaken ex) {
        throw ex;
      }
      catch (Exception e) {
       log.error("NewNode joinFuture interrupted!"+e);
      }
  }

  /** first node should call this **/
  public void create()
  {
    try {
      join(null);
    }
    catch (DKSTooManyRestartJoins ex) {
      log.fatal("Programming bug: creating a ring, but got TooManyRestartJoins in an empty ring");
    }
    catch (DKSRefNoResponse ex) {
    	log.error("Programming bug: creating a ring, but join(null) responded with a DKSRefNoResponse");
    }
    catch (DKSIdentifierAlreadyTaken ex) {
    	log.fatal("Already Taken identifier in an empty ring! Fatal Error!\n"+ex);
    }
  }

  public void join(DKSRef existingnodeAddress ) throws DKSTooManyRestartJoins, DKSIdentifierAlreadyTaken, DKSRefNoResponse
  {

    AsyncOperation joinFuture=null;
    try {
        joinFuture = myDKSNode.newNode( myDKSRef, existingnodeAddress );
        if (joinFuture != null) {
            log.debug("about to wait on future from DKSNOde.newNode()");
            joinFuture.waitOn();
        }
      } catch (DKSTooManyRestartJoins e) {
        throw e;
      } catch (org.kth.dks.dks_exceptions.DKSIdentifierAlreadyTaken ex) {
        throw ex;
      }
      catch (DKSRefNoResponse ex) {
        throw ex;
      }
      catch (Exception e) {
        log.error("NewNode joinFuture interrupted!"+e);
      }
  }

  public void logLevel(int level)
  {
    //DKSPrint.logLevel(level);
  }

  /**
   * Disconnect the local node from the network.
   */
  public void leave() {
    try {
      AsyncOperation endSync = null;
      endSync=myDKSNode.prepareForLeave();
      endSync.waitOn();
    } catch (Exception e) {
    	log.error("Leave() endSync interrupted!");
    }
  }

  /**
   * Finds the responsible node for a certain identifier
   * @param identifier long describing the target identifier whose responsible is to be found
   * @return DKSRef reference containing contact information to the responsible node
   */
  public DKSRef findResponsible(long identifier) {
    DKSRef responsibleNode = null;
    AsyncOperation res = myDKSNode.findResponsible(identifier);
    try {
      responsibleNode = (DKSRef)res.waitOn();
    } catch (Exception ex) {
    	log.error("findResponsible: AsyncOperation was interrupted or cancelled");
    }
    return responsibleNode;
  }

  /**
   * Will route a message the DKS node that is responsible for the given identifier
   * and give it the payload, upon which the responsible node will respond
   * with a payload which is returned
   * @param identifier whose responsible node being routed to
   * @param payload to be sent to the responsible node
   * @return DKSOBject payload sent back by the responsible node
   */
  public DKSObject route(long identifier, DKSObject payload) {
    DKSObject responsePayload = null;
    AsyncOperation res = myDKSNode.route(identifier, payload);
    try {
      responsePayload = (DKSObject)res.waitOn();
    } catch (Exception ex) {
        log.error( "route: AsyncOperation was interrupted or cancelled");
    }
    return responsePayload;
  }

  /**
   * Will route a message to the DKS node that is responsible for the given identifier
   * and give it the payload, upon which the responsible node will respond
   * with a payload which is returned
   * @param identifier whose responsible node being routed to
   * @param payload to be sent to the responsible node
   * @return DKSOBject payload sent back by the responsible node
   */
  public void routeAsync(long identifier, DKSObject payload) {
    myDKSNode.routeAsync(identifier, payload);
  }

  /**
   * Will route a message from firstNode to the DKS node that is responsible for the given identifier
   * and give it the payload, upon which the responsible node will respond
   * with a payload which is returned
   * @param identifier long
   * @param payload DKSObject
   * @param firstNode DKSRef
   */
  public void routeAsyncFrom(long identifier, DKSObject payload, DKSRef firstNode) {
    myDKSNode.routeAsyncFrom(identifier, payload, false, firstNode);
  }

  /**
   * Send the message to the specified recipient node.
   */
  public void send(DKSRef target, DKSMessage message) {
      myDKSNode.send(target,message);
  }

  /**
   * Broadcasts the given message.
   */
  public void broadcast(DKSObject message) {
      myDKSNode.broadCast(message);
  }

  /**
   * Broadcasts to nodes with identifiers within the rand startId and endId []
   * @param message DKSObject message to be sent
   * @param startId long beginning of the interval
   * @param endId long end of the interval
   */
  public void broadcastRestricted(DKSObject message, long startId, long endId) {
    myDKSNode.broadCastRestricted(message, startId, endId);
  }

  /**
   * Instructs the DKS system to pass on all received messages to the
   * specified DKSMessageReceiver.
   */
  public DKSCallbackInterface setCallbackHandler(DKSAppInterface appHandler) {
    return myDKSNode.setCallbackHandler(appHandler);
  }

    public NodeInfo getNodeInfo(){
	return myDKSNode.getNodeInfo();
    }

    public java.io.Serializable getDebugInfo() {
	return myDKSNode.getDebugInfo();
    }


    public CommunicationInfo getComInfo(){
	return myDKSNode.getComInfo();
    }

    public MessageInfo [] getMessageInfo(){
	return myDKSNode.getMessageInfo();
    }

    /* FIXME To be removed. */
    public DKSMarshal getDKSMarshal(){
	return myDKSNode.getDKSMarshal();
    }

    public long addressSpace(){
	return DKSNode.N;
    }

    public DKSNode getMyDKSNode(){
	return myDKSNode;
    }

    public void unregisterNode() {
      cm.unregisterDKSNode(myDKSRef);
    }

    public void joinCallbackReturn() {
      myDKSNode.joinCallbackReturn();
    }

    /**
     * <b>INVARIANT:</b> should be called by DKSAppInterface every time DKSAppInterface.leaveCallback() is called
     * @see DKSAppInterface
     */
    public void leaveCallbackReturn() {
      myDKSNode.leaveCallbackReturn();
    }

    public ConnectionManager getCM() {
	return cm;
    }

    /**
     * @see DKSInterface.addMsgHandler()
     */
    public boolean addMsgHandler(DKSMessage msg, Object handlerObject, String methodName) {
    	return myDKSNode.addMsgHandler(msg, handlerObject, methodName);
    }

    public DKSRef getDKSRef() {
    	return myDKSRef;
    }
    
}//DKSTest class
